package com.icreate.storm.methods.TupleAccumulate.indicator;

import com.icreate.storm.methods.TupleAccumulate.TupleAccumulator;
import com.icreate.entity.indicator.IndicatorID;
import org.apache.storm.tuple.ITuple;
import org.bson.Document;

import java.io.Serializable;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * Created by feng on 2017/5/14.
 */
public class IndicatorStaticAccumulator implements TupleAccumulator {

    private Map<String, IndicatorBuffer> counter = new LinkedHashMap<>();


    public IndicatorStaticAccumulator(Map<String, String[]> indicatorsFields) {

        for (Map.Entry<String, String[]> entry : indicatorsFields.entrySet()) {
            this.counter.put(entry.getKey(), new IndicatorBuffer(entry.getValue()));
        }
    }

    @Override
    public void add(ITuple tuple) {
        String type = tuple.getString(0);
        IndicatorID indicatorID = (IndicatorID) tuple.getValueByField("IndicatorID");
        String hour = indicatorID.getHour();
        this.counter.get(type).add(Integer.valueOf(hour), tuple);
//        Fields fields = tuple.getFields();
//        for (String field : fields) {
//            if (!excludeFields.contains(field))
//
//                add("hour." + hour + "." + field, (Long) tuple.getValueByField(field));
//        }
    }

    @Override
    public Document toDocument() {
        Document document = new Document();
        Document incDocument = new Document();
        for (Map.Entry<String, IndicatorBuffer> entry : this.counter.entrySet()) {
            incDocument.putAll(entry.getValue().toDocument());
        }
        if (incDocument.size() == 0)
            return null;
        document.append("$inc", incDocument);
        return document;
    }

    @Override
    public void reset() {
        for (IndicatorBuffer buffer : this.counter.values()) {
            buffer.reset();
        }
    }

    private static class IndicatorBuffer implements Serializable {
        private String[] fields;
        long[][] hours;

        public IndicatorBuffer(String[] fields) {
            this.fields = fields;
            hours = new long[24][fields.length + 1];
        }

        public void add(int hour, ITuple tuple) {
            long sum = 0;
            long value;
            for (int i = 0; i < this.fields.length; i++) {
                value = (long) tuple.getValueByField(fields[i]);
                hours[hour][i] += value;
                sum += value;
            }
            if (sum > 0)
                hours[hour][this.fields.length] = 1;
        }

        public Document toDocument() {
            Document document = new Document();
            String field;
            long valid;
            long value;
            for (int i = 0; i < hours.length; i++) {
                valid = hours[i][fields.length];
                if (valid == 1) {
                    for (int j = 0; j < hours[i].length - 1; j++) {
                        value = hours[i][j];
                        field = fields[j];
                        if (value > 0)
                            document.append("hour." + i + "." + field, value);
                    }
                }
            }
            return document;
        }

        public void reset() {
            for (int i = 0; i < hours.length; i++) {
                for (int j = 0; j < hours[i].length; j++) {
                    hours[i][j] = 0;
                }
            }
        }

    }
}
